package monitor.work

import akka.actor.{Actor, ActorContext, ActorLogging, ActorRef, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import monitor.config.MonitorConfig
import monitor.master.Master
import monitor.model.{BackRegMsgOK, MessageRegister, MessageRegisterReport}

/**
  * 工作节点
  */
class Worker(rm: RemoteMaster) extends Actor with ActorLogging {

  var workerId:String = _

  var selection: ActorSelection = _

  /**
    * 预先准备
    */
  override def preStart(): Unit = {
    var url = s"akka.tcp://${Master.MASTER_SYSTEM}@${rm.masterHost}:${rm.masterPort}/user/${Master.MASTER_ACTOR_SYSTEM}"

    log.info(url)
    selection = context.actorSelection(url)
    println()
    if (log.isInfoEnabled) log.info(s"首次注册 工作节点 ${url}")

    // 第一次启动分配的机器id
    workerId = MonitorConfig.keyGenerationStrategy.toString

    //     发送首次注册信息
    selection ! MessageRegister(workerId, MonitorConfig.systemInfo._1, MonitorConfig.systemInfo._2, System.currentTimeMillis())
  }

  override def receive: Receive = {
    // 注册成功后 ---> 启动 保活机制
    case BackRegMsgOK => {
      println(s"############ worker 注册成功 工作机器id  ${workerId}  ##############")
      val info = MessageRegisterReport(workerId,System.currentTimeMillis())
      // 启动定时 保活任务
      Worker.reportKepperAlived (context,self,info)
    }
      // 发送注册消息
    case worker: MessageRegisterReport => {
      //
      if (log.isInfoEnabled) {
        log.info("####### 开始向 master 进行健康报告 #######")
        log.info(s"#######  进行健康报告数据 ${worker} #######")
      }
      selection ! worker
    }
  }
}

object Worker {

  val WORKER_SYSTEM = "worker_system"
  val WORKER_ACTOR_SYSTEM = "worker_actor_system"

  /**
    *  initialDelay: FiniteDuration,
        interval:     FiniteDuration,
        receiver:     ActorRef,
        message:      Any)
    *
    * @param ctx
    * @return
    */

  def reportKepperAlived(ctx:ActorContext,receiver: ActorRef,msg:MessageRegisterReport) = {
    println("#############################  准备启动 心跳检测 的任务 #############################")
    import scala.concurrent.duration._
    import ctx.dispatcher
 //   val info = MessageRegisterReport(workerId,System.currentTimeMillis())
    ctx.system.scheduler.schedule(0 second,3 second,receiver,msg)

  }

  def main(args: Array[String]): Unit = {

    println("#############################  启动 Worker 节点 #############################")

    val Array(host, port) = args
    /**
      * 启动
      */
    val s: String =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = "127.0.0.1"
         |akka.remote.netty.tcp.port = 4444
      """.stripMargin

    val config = ConfigFactory.parseString(s)

    // 启动注册
    val workersystem = ActorSystem(WORKER_SYSTEM, config)
    workersystem.actorOf(Props(new Worker(RemoteMaster())), WORKER_ACTOR_SYSTEM)
  }
}


/**
  * 远程的 master 地址
  *
  * @param masterHost
  * @param masterPort
  */
case class RemoteMaster(val masterHost: String = "127.0.0.1", val masterPort: Int = 9999)